package com.rem.zookeeper.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;


/**
 * 手动实现分布式锁 生产不可用
 * 原理： 1.正常获取到锁的情况：创建一个临时有序的节点
 * *     2.获取不到锁的情况：查看父节点下所有节点，获取比当前节点小1的节点并且监听，当比自己小1的节点被移除就获取到锁，其余时间等待
 * （1）会话连接是异步的，需要自己去处理。比如使用 CountDownLatch
 * （2）Watch 需要重复注册，不然就不能生效
 * （3）开发的复杂性还是比较高的
 * （4）不支持多节点删除和创建。需要自己去递归
 *
 * @author Rem
 * @date 2022/12/4
 */
public class DistributedLock {

    private static ZooKeeper zk;
    /**
     * 创建的当前子节点
     */
    private static String currentNode;
    /**
     * 父节点
     */
    private static final String ROOT_NODE = "locks";
    /**
     * 子节点路径
     */
    private static String waitPath;
    /**
     * zk 节点等待
     */
    private static final CountDownLatch WAIT_LATCH = new CountDownLatch(1);


    /**
     * 构造函数 创建连接
     *
     * @throws Exception
     */
    public DistributedLock() {
        try {
            getConnect();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void getConnect() throws IOException, InterruptedException, KeeperException {
        CountDownLatch connectLatch = new CountDownLatch(1);
        zk = new ZooKeeper("localhost:2191,localhost:2192,localhost:2193", 5000, event -> {
            // 连接建立时, 打开 latch, 唤醒 wait 在该 latch 上的线程
            if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                connectLatch.countDown();
            }
            // 发生了 waitPath 的删除事件
            if (event.getType() == Watcher.Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                WAIT_LATCH.countDown();
            }
        });
        connectLatch.await();
        //获取根节点状态
        Stat stat = zk.exists("/" + ROOT_NODE, false);
        //如果根节点不存在，则创建根节点，根节点类型为永久节点
        if (stat == null) {
            zk.create("/" + ROOT_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 加锁
     *
     * @throws Exception
     */
    public void zkLock() throws Exception {
        //在根节点下创建临时顺序节点，返回值为创建的节点路径
        currentNode = zk.create("/" + ROOT_NODE + "/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        List<String> childrenNodes = zk.getChildren("/" + ROOT_NODE, false);
        // 列表中只有一个子节点, 那肯定就是 currentNode , 说明  client 获得锁
        if (childrenNodes.size() > 1) {
            //对根节点下的所有临时顺序节点进行从小到大排序
            Collections.sort(childrenNodes);
            //当前节点名称
            String thisNode = currentNode.substring(("/" + ROOT_NODE + "/").length());
            //获取当前节点的位置
            int index = childrenNodes.indexOf(thisNode);
            if (index == -1) {
                System.out.println("数据异常");
            } else if (index == 0) {
                System.out.println("当前client 获得锁");
            } else {
                // 获得排名比 currentNode 前 1 位的节点
                waitPath = "/" + ROOT_NODE + "/" + childrenNodes.get(index - 1);
                // 在 waitPath 上注册监听器, 当 waitPath 被删除时, zookeeper 会回调监听器的 process 方法
                zk.getData(waitPath, true, new Stat());
                //进入等待锁状态
                WAIT_LATCH.await();
            }
        }
    }

    /**
     * 解锁
     *
     * @throws Exception
     */
    public void zkUnlock() throws Exception {
        zk.delete(currentNode, -1);
    }
}
